package org.nbict.iot.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.nbict.iot.protocol.util.BytesOperator;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.Properties;

/**
 * Created by songseven on 18/6/13.
 */
public class SimpleProducer {

    public static void main(String[] args) {
        // Check arguments length value
        if (args.length == 0) {
            System.out.println("Enter topic name");
            return;
        }

        //Assign topicName to string variable
        String topicName = args[0].toString();

        // create instance for properties to access producer configs   
        Properties props = new Properties();

        //Assign localhost id
        props.put("bootstrap.servers", "localhost:9092");

        //Set acknowledgements for producer requests.      
        props.put("acks", "all");

        //If the request fails, the producer can automatically retry,
        props.put("retries", 0);

        //Specify buffer size in config
        props.put("batch.size", 16384);

        //Reduce the no of requests less than 0   
        props.put("linger.ms", 1);

        //The buffer.memory controls the total amount of memory available to
        // the producer for buffering.
        props.put("buffer.memory", 33554432);

        props.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");

        props.put("value.serializer",
                "org.apache.kafka.common.serialization.ByteArraySerializer");

        Producer<String, byte[]> producer = new KafkaProducer<String, byte[]>
                (props);

//        byte[] raw = BytesOperator.hexStr2Bytes("01232306fe313131313131313131313131313131313101000812060c0f31020006e2");
//        for (int i = 0; i < 20; i++) {
//            producer.send(new ProducerRecord<String, byte[]>(topicName,
//                    Integer.toString(i), raw));
//            Utils.sleep(3000);
//        }
//
//        System.out.println("Message sent successfully");

        File file = new File("/Users/songseven/_logs/export-2018-07-20-11_25_42.log");
        BufferedReader reader = null;

        try {

            reader = new BufferedReader(new FileReader(file));
            String tempString = null;
            int line = 1;
            while ((tempString = reader.readLine()) != null) {
                System.out.println((line++) + " ==> " + tempString);
                byte[] raw = BytesOperator.hexStr2Bytes(tempString);
                producer.send(new ProducerRecord<String, byte[]>(topicName, Integer.toString(line), raw));
                //Utils.sleep(1);
            }
            System.out.println("end");
            reader.close();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (reader != null) {
                try {
                    reader.close();
                } catch (IOException e1) {
                }
            }
        }

        producer.close();
    }
}
